package org.zookeeper.study;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

/** 
* @author James
* @date 2018年8月16日 
*/
public class CreateSession {
	private CuratorFramework client=null;
	
	public CreateSession(CuratorFramework client) {
		this.client=client;
	}
	
	public static void main(String[] args) throws Exception {
		//刚开始重试间隔1秒，之后重试间隔逐渐增加，最多重试不超过三次
		RetryPolicy retryPolicy=new ExponentialBackoffRetry(1000,30);
		CuratorFramework client=CuratorFrameworkFactory.builder().connectString("192.168.25.177:2181")
				.sessionTimeoutMs(3000)
				.connectionTimeoutMs(3000)
				.retryPolicy(retryPolicy)
				.build();
		
		client.start();
		
		CreateSession session=new CreateSession(client);
//		session.createStat("/a/b/c", "321");
//		System.out.println(session.select("/a/b/c"));
//		
////		session.delete("/a");
//		session.updateStat("/a/b/c", "111");
		
		session.asynchCall();
		
		session.eventListen("/node_1");
		
//		String path=client.create().creatingParentsIfNeeded()
//				.withMode(CreateMode.EPHEMERAL)//withMode节点类型
//				.forPath("/curator/3","131".getBytes());
//		System.out.println(path);
//		List<String> list=client.getChildren().forPath("/");
//		System.out.println(list);
//		
//		list.forEach(s->{
//			try {
//				String re=new String(client.getData().forPath("/"+s));
//				System.out.println("key:"+s+",value:"+re);
//			} catch (Exception e) {
//				System.out.println("key:"+s);
//				e.printStackTrace();
//			}
//		});
//		
//		Stat stat=new Stat();
//		String re=new String(client.getData().storingStatIn(stat).forPath("/curator/3"));
//		System.out.println(re);
//		System.out.println(stat);
//		
//		client.delete().guaranteed()//保障机制，若未删除成功，只要会话会在后台一直尝试删除
//			.deletingChildrenIfNeeded()
//			.withVersion(-1)
//			.forPath("/curator");
		Thread.sleep(Integer.MAX_VALUE);
	}
	
	public void createStat(String key,String value) {
		String path;
		try {
			path = client.create().creatingParentsIfNeeded()
					.withMode(CreateMode.EPHEMERAL)//withMode节点类型
					.forPath(key,value.getBytes());
			System.out.println(path);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void delete(String path) {
		try {
			client.delete().guaranteed()//保障机制，若未删除成功，只要会话会在后台一直尝试删除
				.deletingChildrenIfNeeded()
				.withVersion(-1)
				.forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public String select(String key) {
		Stat stat=new Stat();
		try {
			String re=new String(client.getData().storingStatIn(stat).forPath(key));
			return re;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public void updateStat(String key,String value) {
		Stat stat=new Stat();
		try {
			String re=new String(client.getData().storingStatIn(stat)//在获取节点内容的同时把状态信息存入Stat
					.forPath(key));
			System.out.println(re);
			System.out.println(stat);
			Thread.sleep(10000);
			client.setData().withVersion(stat.getVersion())//修改前获取一次节点数据得到版本信息
				.forPath(key,value.getBytes());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public ExecutorService es=Executors.newFixedThreadPool(5);
	
	public void asynchCall() {
		try {
			client.checkExists().inBackground(new BackgroundCallback() {
				@Override
				public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
					System.out.println(event.getType());
					int re=event.getResultCode();//执行成功为0
					System.out.println(re);
					
					String path=event.getPath();
					System.out.println(path);
					Stat stat=event.getStat();
					System.out.println(stat);
				}
			},"123123",es).forPath("/node_1");//把线程池es传给异步调用
			
			List<String> list=client.getChildren().forPath("/");
			System.out.println(list);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void eventListen(String key) {
		NodeCache cache=new NodeCache(client,key);
		try {
			cache.start();
			cache.getListenable().addListener(new NodeCacheListener() {
				@Override
				public void nodeChanged() throws Exception {
					byte[] res=cache.getCurrentData().getData();
					System.out.println("data:"+new String(res));
				}
				
			}, es);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}
}
